package cn.liuwenlin.mq;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;
import java.util.Scanner;
import java.util.concurrent.TimeUnit;

/**
 * activemq数据接受
 * @author liuwenlin
 */
public class Receiver {

    volatile static boolean sentinel = false;

    static Thread forkThread;

    public static void main(String[] args) {

        final Thread mainThread = Thread.currentThread();

        //测试中断
        forkThread = new Thread(new Runnable() {
            @Override
            public void run() {
                Scanner scanner = new Scanner(System.in);
                String inputString;
                while((inputString = scanner.next()) != null){
                    if("stop".equals(inputString)){
                        Receiver.stop(mainThread);
                        break;
                    }
                }
            }
        });

        forkThread.start();
        System.out.println("listener thread started...");

        //1.建立连接工厂
        ActiveMQConnectionFactory activeMQConnectionFactory
                = new ActiveMQConnectionFactory(
                ActiveMQConnectionFactory.DEFAULT_USER,
                ActiveMQConnectionFactory.DEFAULT_PASSWORD,
                "nio://hadoop.master.com:5671");
        Connection connection = null;
        try {
            //2.获取一个向Active MQ的连接
            connection = activeMQConnectionFactory.createConnection();

            //3.获取session
//            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            //4.找目的地，获取destination，消费端，也会从这个目的地取消息
            Destination mq = session.createQueue("activemq");

            //5.消息消费者
            /**
             * 消费者端可以配合生产端的MapMessage设置指定的group组参数，
             * 以达到不同消费端消费不同组别的消息，有点类似于负载均衡的机制
             */
            String selector = "age < 26 and price < 280";

            MessageConsumer consumer = session.createConsumer(mq, selector);

            connection.start();


            //6.接受消息
            while(!Thread.currentThread().isInterrupted()){
//                TextMessage message = (TextMessage)consumer.receive();
//                System.out.println("receive msg:" + message.getText());
                MapMessage msg = (MapMessage)consumer.receive();
                System.out.println("------");
                System.out.println("receive msg: name " + msg.getString("name"));
                System.out.println("receive msg: age " + msg.getInt("age"));
                System.out.println("receive msg: price " + msg.getDouble("price"));
//                message.acknowledge();
                TimeUnit.MILLISECONDS.sleep(100);
            }

        } catch (JMSException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            System.out.println("Main thread has been interrupted...");
            // 重置中断标志位为true
            Thread.currentThread().interrupt();
            e.printStackTrace();
        } finally {

            //6.关闭连接
            if(connection != null){
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }

            System.out.println("exit...");
        }

    }

    public static void stop(Thread thread){
        sentinel = true;
        thread.interrupt();
    }

}
